package com.zhisheng.sql.blink.stream.example;

import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;


/**
 * CasinoWinJob
 *
 * @author yongxiang
 * @date 2025/2/28 14:59
 * @desc
 */
public class CasinoWinJob {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        TableConfig config = tableEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","5s");
        // 定义 Debezium Kafka 源表
        tableEnv.executeSql(
                "CREATE TABLE t_topology ("
                        + "  topology_id INT,"
                        + "  parent_topology_id INT,"
                        + "  name STRING, "
                        + "  status STRING, "
                        + "  topology_type_id INT "
                        + ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.t_topology ',"
                        + "  'properties.bootstrap.servers' = '192.168.50.235:9092,192.168.50.235:9093,192.168.50.235:9094',"
                        + "  'properties.group.id' = 'test_consumer',"
                        + "  'scan.startup.mode' = 'earliest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");

        // 3. 创建过滤视图
        String createFilteredView =
                "CREATE TEMPORARY VIEW t_topology_view AS \n" +
                        "SELECT topology_id, parent_topology_id, name,topology_type_id \n" +
                        "FROM t_topology \n" +
                        "WHERE status = 'ACTIVE' AND topology_type_id IN (300,400,1200)";
        tableEnv.executeSql(createFilteredView);


        // 定义 Debezium Kafka 源表
        tableEnv.executeSql(
                "CREATE TABLE t_bet  ("
                        + "  bet_id INT,"
                        + "  wager DOUBLE,"
                        + "  casino_win DOUBLE, "
                        + "  table_id INT, "
                        + "  payout_complete_dtm BIGINT, " // 原始字段为 BIGINT（Unix 时间戳，毫秒）
                        + "  ts AS TO_TIMESTAMP_LTZ(payout_complete_dtm - 8 * 3600 * 1000, 3), "  // 转换为 TIMESTAMP
                        + "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND"
                        + ") WITH ("
                        + "  'connector' = 'kafka',"
                        + "  'topic' = 'mysql-server.test_sm.t_bet',"
                        + "  'properties.bootstrap.servers' = '192.168.50.235:9092,192.168.50.235:9093,192.168.50.235:9094',"
                        + "  'properties.group.id' = 'click_consumer',"
                        + "  'scan.startup.mode' = 'latest-offset',"
                        + "  'format' = 'debezium-json',"
                        + "  'debezium-json.schema-include' = 'true'"
                        + ")");





        Table result = tableEnv.sqlQuery(
                        "   Select \n" +
                        "    tto.name AS table_name, \n" +
                        "    ttp.name AS pit_name, \n" +
                        "    tth.name AS hall,\n" +
                        "    SUM(tbu.casino_win) AS casino_win, \n" +
                                "  TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start, "+
                                "  TUMBLE_END(ts, INTERVAL '5' MINUTE) AS window_end" +
                        "    FROM t_bet tbu \n" +
                        "    LEFT JOIN t_topology_view tto ON tbu.table_id=tto.topology_id\n" +
                        "    LEFT JOIN t_topology_view ttp ON tto.parent_topology_id = ttp.topology_id\n" +
                        "    LEFT JOIN t_topology_view tth ON ttp.parent_topology_id = tth.topology_id\n" +
                                " GROUP BY " +
                         "  tto.name,ttp.name,tth.name,"+
                         "  TUMBLE(tbu.ts, INTERVAL '5' MINUTE)"
        );


        // 配置 JDBC Sink 将数据实时写入 MySQL
        String mysqlUrl = "jdbc:mysql://192.168.50.235:3306/test_sm";
        String mysqlUser = "root";
        String mysqlPassword = "Cyx1018!";
        // 设置 JDBC Sink
        SinkFunction<Row> jdbcSink = JdbcSink.sink(
                "INSERT INTO result_total (table_name,pit_name, hall, window_start,window_end,casino_win) VALUES (?, ?, ?,?,?,?) " +
                        "ON DUPLICATE KEY UPDATE casino_win = VALUES(casino_win)",
                (ps, t) -> {
                    ps.setString(1, String.valueOf( t.getField(0).toString())); // table_name
                    ps.setString(2, String.valueOf( t.getField(1)));  // pit_name
                    ps.setString(3, String.valueOf( t.getField(2)));  // hall
                    ps.setString(4, String.valueOf(t.getField(4)));  // window_start
                    ps.setString(5, String.valueOf(t.getField(5)));  // window_end
                    ps.setDouble(6, Double.valueOf( t.getField(3).toString())); // casino_win
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(1) // 每条数据单独插入
                        .withMaxRetries(3) // 重试次数
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(mysqlUrl)
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername(mysqlUser)
                        .withPassword(mysqlPassword)
                        .build()
        );
        // 4. 输出中间结果（调试用）
        Table topologyTable = tableEnv.from("t_topology_view");
        tableEnv.toChangelogStream(topologyTable).print("t_topology_view");
        // 4. 输出中间结果（调试用）
        Table betTable = tableEnv.from("t_bet");
        tableEnv.toChangelogStream(betTable).print("t_bet");
        // 将结果转换为 DataStream
        DataStream<Row> resultStream = tableEnv.toChangelogStream(result);
        // 将结果流写入 MySQL
        resultStream.addSink(jdbcSink);
        env.execute("Real-time total....");

    }


}
